package com.hutu.consummer;

import com.hutu.model.Student;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsummerService {
    @KafkaListener(topics = "test-kafka",groupId = "testgroupsss")
    public void sendMessage(@Payload String msg,
                            @Headers MessageHeaders headers,
                            ConsumerRecord<String,Object> record) {
        System.out.println("收到消息：" + msg + "，消息头：" + headers);
        System.out.println("收到消息是：" + record);
    }

    /**
     * 对象类型的
     * @param headers
     * @param record
     */
    @KafkaListener(topics = "test-kafka-user",groupId = "testgroupsssKK")
    public void sendMessage2(@Payload Student student,
                             @Headers MessageHeaders headers,
                             ConsumerRecord<String,Object> record,
                             Acknowledgment acknowledgment) {
        System.out.println("收到消息：" + student.toString() + "，消息头：" + headers);
        System.out.println("收到消息是：" + record);
        // 手动确认
        acknowledgment.acknowledge();
    }

    /**
     * 意思就是: 创建 test-kafka-test topic,然后读取指定分区 {0,1},并且读取 分区 3 里的偏移量3后面的数据 ...
     */
    @KafkaListener(groupId = "test_group",
            topicPartitions = {
                    @TopicPartition(
                            // 指定topic
                            topic = "test-kafka-test",
                            // 指定分区 读取的分区
                            partitions = {"0", "1"},
                            partitionOffsets = {
                                    // 指定读取的分区的偏移量后面的
                                    @PartitionOffset(partition = "3", initialOffset = "3"),
                                    @PartitionOffset(partition = "4", initialOffset = "4")
                            }
                    )
            })
    public void sendMessageGroup(@Payload String msg,
                            @Headers MessageHeaders headers,
                            ConsumerRecord<String,Object> record) {
        System.out.println("收到消息：" + msg + "，消息头：" + headers);
        System.out.println("收到消息是：" + record);
    }
}
